# 作业
# 1、统计所有科目都及格的学生，输出结果【姓名，班级，科目名称，分数】
#
# 2、统计总分超过班级平均分的学生，输出结果【姓名，班级，总分】
#
# 3、统计每个班级总分前十的学生，输出结果【姓名，班级，总分】
#
# 4、统计偏科最严重的前10个学生，输出结果【姓名，班级，科目名称，分数】
from pyspark.sql.session import SparkSession

spark = SparkSession \
    .builder \
    .master("local") \
    .appName("demo25_student") \
    .getOrCreate()

# 获取sparkContext
sc = spark.sparkContext

# 1、统计所有科目都及格的学生，输出结果【姓名，班级，科目名称，分数】

# 读取数据
students_rdd = sc.textFile("../../data/students.txt")
subjects_rdd = sc.textFile("../../data/subject.txt")
scores_rdd = sc.textFile("../../data/score.txt")

# 关联分数表和科目表
score_subject_rdd = scores_rdd \
    .map(lambda line: (line.split(",")[1], line)) \
    .join(subjects_rdd.map(lambda line: (line.split(",")[0], line)))


# ('1000008', ('1500100993,1000008,7', '1000008,地理,100'))
def score_subject_fun(kv):
    # 学号
    id = kv[1][0].split(",")[0]
    # 科目名称
    subject_name = kv[1][1].split(",")[1]
    # 分数
    score = int(kv[1][0].split(",")[-1])
    return id, (subject_name, score)


# ('1500101000', (('语文', 100), '1500101000,符瑞渊,23,男,理科六班'))
def student_score_fun(kv):
    id = kv[0]
    subject_name = kv[1][0][0]
    score = kv[1][0][1]
    name = kv[1][1].split(",")[1]
    clazz = kv[1][1].split(",")[-1]
    return id, name, clazz, subject_name, score


# 关联学生表
student_score_rdd = score_subject_rdd \
    .map(score_subject_fun) \
    .join(students_rdd.map(lambda line: (line.split(",")[0], line))) \
    .map(student_score_fun)


# 判断是否及格
# ('1000008', ('1500100959,1000008,99', '1000008,地理,100'))
def filter_fun(kv):
    # 学生的分数
    score = int(kv[1][0].split(",")[-1])
    # 科目的总分
    sum_score = int(kv[1][1].split(",")[-1])
    # 判断是否及格
    return score >= sum_score * 0.6


score_filter_rdd = score_subject_rdd.filter(filter_fun)

# 计算每个学生及格的科目数量,取出都及格的学生
# ('1000008', ('1500100862,1000008,90', '1000008,地理,100'))
pass_ids = score_filter_rdd \
    .map(lambda kv: (kv[1][0].split(",")[0], 1)) \
    .reduceByKey(lambda a, b: a + b) \
    .filter(lambda kv: kv[1] == 6) \
    .map(lambda kv: kv[0]) \
    .collect()

# 取出所有科目都及格的学生的分数
pass_rdd = student_score_rdd.filter(lambda info: info[0] in pass_ids)

# pass_rdd.foreach(print)


# 4、统计偏科最严重的前10个学生，输出结果【姓名，班级，科目名称，分数】
# 取出学号和分数,按照学号进行分组
score_group_by_rdd = scores_rdd \
    .map(lambda line: (line.split(",")[0], int(line.split(",")[-1]))) \
    .groupByKey()


# 计算学生分数的方差
def std_fun(kv):
    id = kv[0]
    scores = kv[1]
    n = len(scores)
    avg = sum(scores) / n
    # 计算方差公式
    std = sum([(score - avg) ** 2 for score in scores]) / n
    return id, std


# 计算学生分数的方差,按照方差排序，取前十
top10 = score_group_by_rdd \
    .map(std_fun) \
    .sortBy(lambda kv: kv[1], ascending=True) \
    .map(lambda kv: kv[0]) \
    .take(10)

# 获取偏科严重学生分数
student_score_rdd.filter(lambda info: info[0] in top10).foreach(print)
